package com.zjl.Kafka.第03章_操作kafka;


import org.apache.kafka.clients.admin.*;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * **** zookeeper 启动****************************************************************
 * source /etc/profile
 * cd /www/kafka/kafka_2.12-3.8.0/bin/
 * sh zookeeper-server-start.sh   ../config/zookeeper.properties
 *
 * ****单台 kafka 启动****************************************************************
 * source /etc/profile
 * cd /www/kafka/kafka_2.12-3.8.0/bin/
 * sh kafka-server-start.sh  ../config/server.properties
 *
 * ****集群 kafka 启动****************************************************************
 * source /etc/profile
 * cd /www/kafka/kafka-1/bin/
 * sh kafka-server-start.sh  ../config/server-1.properties
 *
 * ****集群 kafka 启动****************************************************************
 * source /etc/profile
 * cd /www/kafka/kafka-2/bin/
 * sh kafka-server-start.sh  ../config/server-2.properties
 *
 * ****集群 kafka 启动****************************************************************
 * source /etc/profile
 * cd /www/kafka/kafka-3/bin/
 * sh kafka-server-start.sh  ../config/server-3.properties
 */
public class A_主题创建 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String,Object> confMap = new HashMap<>();
        confMap.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9091,82.157.71.243:9092,82.157.71.243:9093");
        confMap.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9092");

        Admin admin = Admin.create(confMap);//管理员对象
//        AdminClient admin = AdminClient.create(confMap);//管理员对象


        String topicName1 = "test1";//主题的名称   可用字符  [a-zA-Z0-9._|]
        int partitionCount1 = 1;//主题分区数量
        short replicationCount1 = 1;//主题分区副本因子数量
        NewTopic topic1 = new NewTopic(topicName1,partitionCount1,replicationCount1);

        String topicName2 = "test2";//主题的名称   可用字符  [a-zA-Z0-9._|]
        int partitionCount2 = 2;//主题分区数量
        short replicationCount2 = 2;//主题分区副本因子数量    要小于等于 kafka 节点个数
        NewTopic topic2 = new NewTopic(topicName2,partitionCount2,replicationCount2);

        String topicName4 = "test4";//主题的名称   可用字符  [a-zA-Z0-9._|]
        Map<Integer, List<Integer>> map = new HashMap<>();
        map.put(0,Arrays.asList(1,2));
        map.put(1,Arrays.asList(2,3));
        map.put(2,Arrays.asList(3,1));
        NewTopic topic4 = new NewTopic(topicName4,map);

        //创建主题
        CreateTopicsResult topics = admin.createTopics(Arrays.asList(topic4));

        //睡眠的目的是  保证数据在传入kafka后再退出
        try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { throw new RuntimeException(e); }
        admin.close();//关闭管理者

    }
}
